<?php

declare(strict_types=1);

namespace App\Task;

use App\Exception\MessageException;
use Hyperf\Task\Annotation\Task;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Command;
use MongoDB\Driver\Exception\Exception;
use MongoDB\Driver\Manager;
use MongoDB\Driver\Query;
use MongoDB\Driver\WriteConcern;
use function Hyperf\Config\config;

class MongoTask
{
    private ?Manager $manager = null;
    private string $dbName = 'seeklove';

    /**
     * 将object转化为array
     * @param $data
     * @return array
     */
    public function objectToArray($data)
    {
        $results = array();
        foreach ($data as $v) {
            if (is_object($v)) {
                $v = (array)$v;
            }
            $results[] = $v;
        }
        return $results;
    }

    #[Task]
    public function insert(string $namespace, array $document)
    {
        $writeConcern = new WriteConcern(WriteConcern::MAJORITY, 1000);
        $bulk = new BulkWrite();
        $bulk->insert($document);

        // $namespace = $this->dbName . '.' . $namespace;
        $result = $this->manager()->executeBulkWrite($this->dbName . '.' . $namespace, $bulk, $writeConcern);
        return $result->getInsertedCount();
    }

    #[Task]
    public function find(string $namespace, array $filter = [], array $options = [])
    {
        if (!is_array($filter)) {
            throw new MessageException('filter is not array');
        }

        $query = new Query($filter, $options);
        // $namespace = $this->dbName . '.' . $namespace;
        $cursor = $this->manager()->executeQuery($this->dbName . '.' . $namespace, $query);
        return $this->objectToArray($cursor);
    }


    #[Task]
    public function query(array $filter = [])
    {
        if (empty($filter['aggregate'])) {
            throw new MessageException('aggregate is empty');
        }
        $command = new Command($filter);
        $cursor = $this->manager()->executeCommand($this->dbName, $command);
        return $cursor->toArray();
    }

    /**
     * 更新一条记录
     * @param string $namespace 集合名称
     * @param array $filter 更新的过滤条件
     * @param array $update 更新操作
     * @param bool $multi 是否支持更新多条记录，默认false
     * @param bool $upsert 是否启用未匹配时插入该条记录
     * @return bool|int    返回更新数目（2.4版本不能返回数据）
     */
    #[Task]
    public function update(string $namespace, array $filter, array $update, $multi, $upsert)
    {
        $writeConcern = new WriteConcern(WriteConcern::MAJORITY, 1000);
        $bulk = new BulkWrite();
        $bulk->update($filter, $update, ['multi' => $multi, 'upsert' => $upsert]);

        // $namespace = $this->dbName . '.' . $namespace;
        $result = $this->manager()->executeBulkWrite($this->dbName . '.' . $namespace, $bulk, $writeConcern);
        return $result->getModifiedCount();
    }

    #[Task]
    public function delete(string $namespace, array $filter, array $options = [])
    {
        $writeConcern = new WriteConcern(WriteConcern::MAJORITY, 1000);
        $bulk = new BulkWrite();
        $bulk->delete($filter, $options);

        // $namespace = $this->dbName . '.' . $namespace;
        $result = $this->manager()->executeBulkWrite($this->dbName . '.' . $namespace, $bulk, $writeConcern);
        return $result->getDeletedCount();
    }

    #[Task]
    public function getCount($namespace, $filter)
    {
        // $cursor = $this->find($namespace, $filter);
        // return count($cursor);
        $command = new Command([
            'count' => $namespace,
            'query' => $filter,
        ]);
        $result = $this->manager()->executeCommand($this->dbName, $command);
        $response = current($result->toArray());
        if ($response->ok == 1) {
            return $response->n;
        }
        return 0;
    }

    protected function manager()
    {
        $mongoDBConfigArray = config('mongodb');
        $this->dbName = $mongoDBConfigArray['mongodb_name'];
        if ($this->manager instanceof Manager) {
            return $this->manager;
        }
        $uri = $mongoDBConfigArray['mongodb_uri'];
        $options = ['connectTimeoutMS' => 50000, 'socketTimeoutMS' => 50000];
        return $this->manager = new Manager($uri, $options);
    }
}
